﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Diagnostics.Contracts;
using System.Fabric;
using System.Net.Sockets;
using System.Threading.Tasks;

namespace System.Threading.Actors
{
    public class FabricServerClient
    {
        public enum FabricResolveReturnCode { Success, Retriable, Fatal }
        // if using connect timeout, use 3 seconds
        private const int connectTimeout = 3000;

        // 2 seconds is too low for a cluster hosted in Azure.  10 sec works reasonably well, but 
        // I was still seeing OperationCanceledExceptions in Azure.
        private const int ResolveTimeoutSeconds = 15;

        private FabricClient _fabricClient;   // Note: All methods accessible from FabricClient are threadsafe.
        private String[] _connectionStrings;  // In case we need to regenerate the FabricClient instance, because we detect it has stale info.

        // Cache FabricClient instances for performance.  We only cache one currently.  If we need to expand,
        // consider using something like a hybrid dictionary optimized for a very small number of elements.
        private static readonly Object s_cacheLock = new Object();
        private static String s_FabricClientCacheAddress = null;
        private static FabricClient s_FabricClientCacheInstance = null;

        public FabricServerClient(string[] connectionStrings)
        {
            Contract.Requires(connectionStrings != null);
            // Number of connection strings will be 0 when running inside the cluster.

            _connectionStrings = connectionStrings;
            _fabricClient = InitializeFabricClient(connectionStrings);
        }

        // This supports mapping ActorFx partition names to WinFab service partition keys, using their Int64 partition scheme.
        // Ignoring the ability to use strings as partitions - they're more for a fixed set of partitions upfront, 
        // like actorA through actorZ, or for state names (Alaska, Illinois, etc).  Instead, we'll hash strings ourselves
        // and use the actor runtime's support for partitioning based on integers.  It maps an Int64 to some number of partitions
        // by subdividing the range of possible values we specify.  It does NOT do a mod operation.
        public static long GetHashedPartitionKey(String partitionName)
        {
            Contract.Requires(partitionName != null);

            long hashedPartitionKey = HardCodedHashFunction(partitionName);
            return hashedPartitionKey;
        }

        // This hash function is intended to be shared across all actor clients and servers, and most likely 
        // across all versions as well.  Never change it once we have shipped it.  Also, use this hard-coded 
        // hash function instead of one in mscorlib, because the BCL is free to reimplement String.GetHashCode
        // daily, and added randomized hashing in .NET 4.5 to work around a security issue.
        // This is the same hash function I used in the ResourceManager's .resources file format, and is a 
        // slow version of the non-randomized String.GetHashCode() we used from .NET 2.0 through at least 4.5.
        private static int HardCodedHashFunction(String key)
        {
            // Never change this hash function.
            uint hash = 5381;
            for (int i = 0; i < key.Length; i++)
                hash = ((hash << 5) + hash) ^ key[i];
            return (int)hash;
        }

        // Return a tuple containing the new TcpClient, the resolved service partition, a unique discriminator for 
        // this service instance, and a return code (ie, success, retriable, etc).
        public async Task<Tuple<TcpClient, ResolvedServicePartition, string, FabricResolveReturnCode>> TryResolveAndCreateRawClientAsync(
            Uri serviceName,
            long? hashedPartitionKey,
            ResolvedServicePartition previousResolvedServicePartition)
        {
            // Note about the reliability of this method: it is entirely possible that a service instance will get load
            // balanced while we are trying to connect to it.  It has been shown to take 16 seconds to fully load balance
            // a service, and during that time we may resolve the service to an endpoint that is no longer listening.
            // So we can easily resolve a service repeatedly, only to find that we cannot connect to it.
            // Let's assume we need to wait a full 30 seconds before giving up.  Retry logic should generally
            // be handled by the caller of this method.
            TcpClient client = null;
            ResolvedServicePartition currentResolvedServicePartition = await ResolveServiceHelperAsync(serviceName, hashedPartitionKey, previousResolvedServicePartition).ConfigureAwait(continueOnCapturedContext:false);
            string uniquePathInfo = null;
            if (currentResolvedServicePartition != null)
            {
                // If we have a previous resolved service partition, we are probably recovering from disconnect
                // When this happens and the current ResolvedServicePartition is the same, then
                // the new address is not available yet. Using the old address to connect could result in a connect timeout.
                // We use a short connect timeout in this case to avoid extra long delays on recovery.
                bool useConnectTimeout = false;
                if (previousResolvedServicePartition != null &&
                    currentResolvedServicePartition.GetEndpoint().Address == previousResolvedServicePartition.GetEndpoint().Address)
                {
                    useConnectTimeout = true;
                }

                // It's possible that the service could "shift" between resolution and raw client generation
                // in a highly volatile environment, and this could result in a socket exception
                // ("connection actively refused").  If there are any problems of this ilk, then behave as
                // if service resolution failed.
                try
                {
                    var tuple = await GenerateRawClientAsync(currentResolvedServicePartition, useConnectTimeout).ConfigureAwait(continueOnCapturedContext:false);
                    client = tuple.Item1;
                    uniquePathInfo = tuple.Item2;
                    return Tuple.Create(client, currentResolvedServicePartition, uniquePathInfo, FabricResolveReturnCode.Success);
                }
                catch (SocketException)
                {
                    // This is expected when a service instance is being load balanced.  We may get essentially incorrect data for
                    // 16 seconds (and probably longer under load).
                    //Console.WriteLine("FabricServerClient::TryResolveAndCreateRawClientAsync - Creating a raw client threw a SocketException.  "+
                    //    "Service: {0} Time: {1}  Message: {2}", serviceName, DateTime.Now, e.Message);
                    client = null;
                    return Tuple.Create(client, currentResolvedServicePartition, uniquePathInfo, FabricResolveReturnCode.Retriable);
                }
            }

            //Console.WriteLine("FabricServerClient::TryResolveAndCreateRawClientAsync - ResolveServiceHelperAsync returned null.");
            return Tuple.Create(client, currentResolvedServicePartition, uniquePathInfo, FabricResolveReturnCode.Fatal);
        }

        private static readonly SemaphoreSlim s_fabricClientSem = new SemaphoreSlim(1, 1);

        private async Task<ResolvedServicePartition> ResolveServiceHelperAsync(Uri serviceName, long? hashedPartitionKey, ResolvedServicePartition previousResolvedServicePartition)
        {
            // Use the Windows Fabric Client to resolve the locations of the actual service instances that this client 
            // needs to talk to.  Passing in the name that the service is associated with and (optionally) a partition 
            // key that can be used to identify a specific partition.  If there's a matching service and partition in the 
            // system, the result will be a collection of information about all of the services that make up the 
            // matching partition.
            ResolvedServicePartition rspIn = previousResolvedServicePartition;
            ResolvedServicePartition rspOut = null;
            bool firstRun = true;
            int retriesLeft = 10;
            while (true)
            {
                if (!firstRun)
                {
                    if (--retriesLeft > 0)
                    {
                        // Resolution failed.  While the service could be failing over, it could be that resolution got flaky.
                        // The experts suggest not throwing away the FabricClient instance though - perhaps one node of data
                        // is obsolete, but the other remaining nodes worth are probably good (perhaps 20% stale, 80% up to date)
                        // Instead pass in the previous resolved service partition, and hope that they refresh their cache.
                        CachedValuesAreInvalid();
                        await Task.Delay(500).ConfigureAwait(continueOnCapturedContext: false);
                    }
                    else
                    {
                        Console.WriteLine("FSC::ResolveServiceHelperAsync - exhausted retry attempts at calling ResolveServicePartitionAsync.");
                        return null;
                    }
                    //Console.WriteLine("Do you want to retry ResolveService? [y/n]");
                    //var line = Console.ReadLine();
                    //if (line != null && !line.Equals("y", StringComparison.CurrentCultureIgnoreCase))
                    //{
                    //    return null;
                    //}
                }
                firstRun = false;
                try
                {
                    Task<ResolvedServicePartition> task = null;

                    try
                    {
                        await s_fabricClientSem.WaitAsync().ConfigureAwait(continueOnCapturedContext: false); // protect access to probably-shared fabricClient
                        if (!hashedPartitionKey.HasValue)
                        {
                            task = _fabricClient.ServiceManager.ResolveServicePartitionAsync(serviceName, rspIn, TimeSpan.FromSeconds(ResolveTimeoutSeconds));
                        }
                        else
                        {
                            task = _fabricClient.ServiceManager.ResolveServicePartitionAsync(serviceName, hashedPartitionKey.Value, rspIn, TimeSpan.FromSeconds(ResolveTimeoutSeconds));
                        }
                    }
                    finally { s_fabricClientSem.Release(); }

                    rspOut = await task.ConfigureAwait(continueOnCapturedContext: false);
                    // Warning: Do not compare the new resolved service partition with the old one, using 
                    // ResolveServicePartition.CompareVersion.  If we resolved to a new service instance (ie, due to failover),
                    // CompareVersion will throw a FabricException saying the metadata of the service is different.
                    //Console.WriteLine("Service Resolved: {0}", rspOut.GetEndpoint().Address);
                    return rspOut;
                }
                catch (AggregateException e)
                {
                    if (e.InnerException is TimeoutException)
                    {
                        Console.WriteLine("Resolve request to (\"{0}\") timed out: RETRIABLE error code", serviceName);
                        continue;
                    }
                    else if (e.InnerException is FabricElementNotFoundException)
                    {
                        // FabricElementNotFoundException indicates that the name in the argument did not exist in the Naming name tree.
                        Console.WriteLine("Name (\"{0}\") does not exists: FATAL error code", serviceName);
                        return null;
                    }
                    else if (e.InnerException is FabricTransientException)
                    {
                        Console.WriteLine("Service (\"{0}\") is unstable: RETRIABLE error code", serviceName);
                        continue;
                    }
                    Console.WriteLine("FATAL exception wrapped in an AggregateException: {0}: {1}", e.InnerException.GetType(), e.InnerException.Message);
                    return null;
                }
                catch (TimeoutException)
                {
                    Console.WriteLine("Resolve request to (\"{0}\") timed out: RETRIABLE error code", serviceName);
                    continue;
                }
                catch (OperationCanceledException)
                {
                    Console.WriteLine("Resolve request to (\"{0}\") was cancelled (timeout?): RETRIABLE error code", serviceName);
                    continue;
                }
                catch (FabricTransientException fte)
                {
                    FabricErrorCode errorCode = fte.ErrorCode;
                    Console.WriteLine("Service (\"{0}\") is unstable: RETRIABLE.  Error code: {1}", serviceName, errorCode);
                    continue;
                }
                catch (ArgumentException e)
                {
                    // One of the common reasons for ArgumentException is that the specified URI is not a valid Windows Fabric name.
                    Console.WriteLine("FATAL exception: {0}: {1}", e.GetType(), e.Message);
                    return null;
                }
                catch (FabricElementNotFoundException)
                {
                    // Service doesn't exist.
                    return null;
                }
                catch (FabricException fe)
                {
                    Console.WriteLine("Fabric exception: {0}.  Error code: {1}", fe.Message, fe.ErrorCode);
                }
                catch (Exception e)
                {
                    Console.WriteLine("FATAL exception: {0}: {1}", e.GetType(), e.Message);
                    return null;
                }
            }
        }

        private static async Task<Tuple<TcpClient,string>> GenerateRawClientAsync(ResolvedServicePartition resolvedServicePartition, bool useConnectTimeout)
        {
            ResolvedServiceEndpoint resolvedServiceEndpoint = resolvedServicePartition.GetEndpoint();
            string listenUri = resolvedServiceEndpoint.Address;
            Uri uri = new Uri(listenUri);

            var uniquePathInfo = uri.LocalPath;
            TcpClient tcp = new TcpClient();
            var connectTask = tcp.ConnectAsync(uri.Host, uri.Port);
            if (useConnectTimeout)
            {
                var firstCompletedTask = await Task.WhenAny(connectTask, Task.Delay(connectTimeout)).ConfigureAwait(continueOnCapturedContext:false);
                if (firstCompletedTask != connectTask)
                {
                    tcp.Close();
                    throw new SocketException();
                }
            }
            else await connectTask.ConfigureAwait(continueOnCapturedContext:false);

            return Tuple.Create(tcp, uniquePathInfo);
        }

        #region FabricClient cache

        // Create a Windows Fabric Client to connect to a Fabric Node that is part of the Windows Fabric deployment.
        // The connection strings that we pass are the Fabric Node listen addresses, which are the primary entry point
        // into the system.  Providing multiple connection strings ensures that if the node that we are talking
        // to goes down (as it would in a node failure) that the clients will automatically connect to another node
        // in the fabric.
        private static FabricClient InitializeFabricClient(string[] connectionStrings)
        {
            //Console.Write("Creating FabricClient ...");
            FabricClient fabricClient = null;

            try
            {
                if (connectionStrings.Length > 1)
                    fabricClient = new FabricClient(connectionStrings);
                else
                {
                    fabricClient = connectionStrings.Length > 0 ? GetFabricClient(connectionStrings[0]) : new FabricClient();
                }
            }
            catch (Exception e)
            {
                Console.WriteLine("Exception caught creating the FabricClient.");
                Console.WriteLine(e);
                return null;
            }

            //Console.WriteLine("Done.");
            return fabricClient;
        }

        [Obsolete("Actor runtime experts claim this is unnecessary.  Stress tests show otherwise.  Try not using this & working with the experts.")]
        internal void CachedValuesAreInvalid()
        {
            // TODO: Ideally FabricClient doesn't need to be reallocated, and we have a simple method to take care of this.
            // Need to chat with Actor Runtime partners and understand what events they provide to us, or vice versa.
            _fabricClient = InitializeFabricClient(_connectionStrings);

            // Dump Fabric Client Cache
            lock (s_cacheLock)
            {
                s_FabricClientCacheInstance = null;
                s_FabricClientCacheAddress = null;
            }
        }

        // Testing with a gateway strongly suggests I need something like this for use within the actor gateway code.
        public static void DumpCachedFabricClient()
        {
            // Dump Fabric Client Cache
            lock (s_cacheLock)
            {
                s_FabricClientCacheInstance = null;
                s_FabricClientCacheAddress = null;
            }
        }

        // WinFab says caching FabricClients is useful.  Our most reliable parallel partitioning test shows a 30% improvement (locally).
        // fabricAddress can be null - if so, we are within the cluster.
        internal static FabricClient GetFabricClient(string fabricAddress)
        {
            Contract.Ensures(Contract.Result<FabricClient>() != null);

            FabricClient fabricClient = null;

            if (fabricAddress == s_FabricClientCacheAddress && s_FabricClientCacheInstance != null)
                fabricClient = s_FabricClientCacheInstance;
            else
            {
                lock (s_cacheLock)
                {
                    if (fabricAddress == s_FabricClientCacheAddress && s_FabricClientCacheInstance != null)
                        fabricClient = s_FabricClientCacheInstance;
                    else
                    {
                        fabricClient = (fabricAddress == null) ? new FabricClient() : new FabricClient(fabricAddress);
                        if (s_FabricClientCacheInstance == null)
                        {
                            s_FabricClientCacheInstance = fabricClient;
                            Thread.MemoryBarrier();
                            s_FabricClientCacheAddress = fabricAddress;
                        }
                        else
                        {
                            Console.WriteLine("Missed a cache lookup for a FabricClient.  Are you connecting to two different fabric clusters?  Cached address: {0}  Current address: {1}", s_FabricClientCacheAddress, fabricAddress);
                            Contract.Assert(s_FabricClientCacheAddress != fabricAddress, "Race in GetFabricClient - missed a cache lookup for the same fabric address?");
                        }
                    }
                }
            }

            return fabricClient;
        }

    #endregion FabricClient cache

    }
}
